package com.tzy.mydb.backend.dm.pageCache;/**
 * @description:
 * @Author HP
 * @create 2022/2/26 16:08
 */

import com.tzy.mydb.backend.common.AbstractCache;
import com.tzy.mydb.backend.dm.page.Page;
import com.tzy.mydb.backend.dm.page.PageImpl;
import com.tzy.mydb.backend.utils.Panic;
import com.tzy.mydb.common.Error;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


/**
 * 页面缓存的具体实现类
 * @author HP
 * @create 2022-02-26 16:08
 * @desc
 **/
public class PageCacheImpl extends AbstractCache<Page> implements PageCache {

    private static final int MEM_MIN_LIM = 10;
    public static final String DB_SUFFIX = ".db";

    private RandomAccessFile file;
    private FileChannel fc;
    private Lock fileLock;

    private AtomicInteger pageNumbers;//记录了当前打开的数据库文件有多少页

    PageCacheImpl(RandomAccessFile file, FileChannel fileChannel, int maxResource) {
        super(maxResource);
        if(maxResource < MEM_MIN_LIM) {
            Panic.panic(Error.MemTooSmallException);
        }
        long length = 0;
        try {
            length = file.length();
        } catch (IOException e) {
            Panic.panic(e);
        }
        this.file = file;
        this.fc = fileChannel;
        this.fileLock = new ReentrantLock();
        this.pageNumbers = new AtomicInteger((int)length / PAGE_SIZE);
    }

    @Override
    /**
     * 数据源就是文件系统， getForCache() 直接从文件中读取，并包裹成 Page
     */
    protected Page getForCache(long key) throws Exception {
        int pgno = (int)key;
        //获得页号的偏移量
        long offset = PageCacheImpl.pageOffset(pgno);

        ByteBuffer buf = ByteBuffer.allocate(PAGE_SIZE);
        fileLock.lock();
        try {
            //读取文件
            fc.position(offset);
            fc.read(buf);
        } catch(IOException e) {
            Panic.panic(e);
        }
        fileLock.unlock();
        return new PageImpl(pgno, buf.array(), this);
    }

    /**
     * // 页号从 1 开始
     * @param pgno
     * @return
     */
    private static long pageOffset(int pgno) {
        return (pgno-1) * PAGE_SIZE;
    }

    @Override
    /**
     * 驱逐页面时，也只需要根据页面是否是脏页面，来决定是否需要写回文件系统
     */
    protected void releaseForCache(Page pg) {
        if(pg.isDirty()) {
            flush(pg);
            pg.setDirty(false);
        }
    }

    /**
     * 写回文件
     * @param pg
     */
    private void flush(Page pg) {
        int pgno = pg.getPageNumber();
        long offset = pageOffset(pgno);

        fileLock.lock();
        try {
            ByteBuffer buf = ByteBuffer.wrap(pg.getData());
            fc.position(offset);
            fc.write(buf);
            fc.force(false);
        } catch(IOException e) {
            Panic.panic(e);
        } finally {
            fileLock.unlock();
        }
    }

    @Override
    /**
     * 新建一个数据页并写入数据库文件
     * pageNumbers这个数字在数据库文件被打开时就会被计算，并在新建页面时自增。
     * 同一条数据是不允许跨页存储的。这意味着，单条数据的大小不能超过数据库页面的大小。
     */
    public int newPage(byte[] initData) {
        int pgno = pageNumbers.incrementAndGet();
        Page pg = new PageImpl(pgno, initData, null);
        flush(pg);
        return pgno;
    }

    @Override
    /**
     * 从缓存中获取指定的数据页
     */
    public Page getPage(int pgno) throws Exception {
        return get((long)pgno);
    }

    @Override
    public void close() {
        super.close();
        try {
            fc.close();
            file.close();
        } catch (IOException e) {
            Panic.panic(e);
        }
    }

    @Override
    public void release(Page page) {
        release((long)page.getPageNumber());
    }

    @Override
    /**
     * 删除指定位置后面的数据页
     */
    public void truncateByBgno(int maxPgno) {
        long size = pageOffset(maxPgno + 1);
        try {
            file.setLength(size);
        } catch (IOException e) {
            Panic.panic(e);
        }
        pageNumbers.set(maxPgno);
    }

    @Override
    /**
     * 获取数据库中的数据页总数
     */
    public int getPageNumber() {
        return pageNumbers.intValue();
    }

    @Override
    public void flushPage(Page pg) {
        flush(pg);
    }
}
